package com.eshipenet.shipnet.dp_decompressor

import org.json4s._
import org.json4s.native.JsonMethods._

import scala.util.Try

object EdgeChannelDataPacket {

  implicit val formats: Formats = DefaultFormats

  def parseMsg(msg: EdgeChannelDataPacketKafkaMessage): Try[EdgeChannelDataPacket] =
    Try {
      val id               = msg.key.split("/", 2)(1)
      val json             = parse(msg.value)
      val data             = (json \ "data").extract[String]
      val compressor       = (json \ "compressor").extract[String]
      val compressorParams = (json \ "compressorParams").extract[Option[JValue]]
      val pFields          = List("compressor", "compressorParams", "data")
      val aux              = json.filterField(f => !pFields.contains(f._1))
      val obj              = JObject(aux)
      EdgeChannelDataPacket(id, data, compressor, compressorParams, obj)
    }

}

case class EdgeChannelDataPacket(id: String,
                                 data: String,
                                 compressor: String,
                                 compressorParams: Option[JValue],
                                 aux: JObject)
